package dev.devlok.springcloudstreamzeromqbinder;

import dev.devlok.springcloudstreamzeromqbinder.producers.ZeroMqMessageProducer;
import dev.devlok.springcloudstreamzeromqbinder.provisioners.ZeroMqMessageBinderProvisioner;

import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
import org.springframework.cloud.stream.binder.ConsumerProperties;
import org.springframework.cloud.stream.binder.ProducerProperties;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.cloud.stream.provisioning.ProducerDestination;
import org.springframework.integration.core.MessageProducer;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;

public class ZeroMqMessageBinder extends AbstractMessageChannelBinder<ConsumerProperties, ProducerProperties, ZeroMqMessageBinderProvisioner> {

    private ZContext zContext;
    private ZMQ.Socket socket;
    private final String socketAddress = "tcp://localhost:6002";

    public ZeroMqMessageBinder(
            String[] headersToEmbed,
            ZeroMqMessageBinderProvisioner provisioningProvider) {

        super(headersToEmbed, provisioningProvider);
    }

    @Override
    protected MessageHandler createProducerMessageHandler(
            final ProducerDestination destination,
            final ProducerProperties producerProperties,
            final MessageChannel errorChannel) throws Exception {

        zContext = new ZContext();
        socket = zContext.createSocket(SocketType.PUB);
        socket.bind(socketAddress);

        return message -> {
            String topic = destination.getName();
            String payload = new String((byte[])message.getPayload());

            socket.sendMore(topic);
            socket.send(payload);
        };
    }

    @Override
    protected void afterUnbindProducer(ProducerDestination destination, ProducerProperties producerProperties) {
        socket.unbind(socketAddress);
        zContext.destroy();

        super.afterUnbindProducer(destination, producerProperties);
    }

    @Override
    protected MessageProducer createConsumerEndpoint(
            final ConsumerDestination destination,
            final String group,
            final ConsumerProperties properties) throws Exception {

        return new ZeroMqMessageProducer(destination);
    }
}
